package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.TableProcess;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.DimUtil;
import com.atguigu.gmall.realtime.utils.HbaseUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;

/**
 * @author Felix
 * @date 2023/11/8
 * 将维度流数据写到Hbase中
 */
public class DimSinkFunction extends RichSinkFunction<JSONObject> {

    private Connection hbaseConn;
    @Override
    public void open(Configuration parameters) throws Exception {
        hbaseConn = HbaseUtil.getHbaseConnection();
    }

    @Override
    public void close() throws Exception {
        HbaseUtil.closeHbaseConnection(hbaseConn);
    }

    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {

        // {"tm_name":"Redmi","id":1,"type":"update","table_process":{"op":"r","sinkColumns":"id,tm_name","sinkFamily":"info","sinkRowKey":"id","sinkTable":"dim_base_trademark","sourceTable":"base_trademark"}}
        //获取对业务数据库维度表数据的操作类型
        //delete:删除表中数据     insert/update/bootstrap-insert:put
        String type = jsonObj.getString("type");
        //获取配置信息
        TableProcess tableProcess = jsonObj.getObject("table_process", TableProcess.class);
        //获取操作的hbase表名称
        String sinkTable = tableProcess.getSinkTable();
        //获取rowkey  id
        String sinkRowKey = tableProcess.getSinkRowKey();
        String rowKeyValue = jsonObj.getString(sinkRowKey);
        if("delete".equals(type)){
            //从hbase表中将对应的记录删除掉
            System.out.println("从hbase的"+sinkTable+"维度表中删除"+rowKeyValue+"数据");
            HbaseUtil.delRow(hbaseConn, GmallConfig.HBASE_NAMESPACE,sinkTable,rowKeyValue);
        }else{
            //向hbase表中put数据
            System.out.println("向hbase的"+sinkTable+"表中插入或者更新"+rowKeyValue+"数据");
            String sinkFamily = tableProcess.getSinkFamily();
            String[] columnNames = tableProcess.getSinkColumns().split(",");
            String[] columnValues = new String[columnNames.length];
            for (int i = 0; i < columnValues.length; i++) {
                columnValues[i] = jsonObj.getString(columnNames[i]);
            }
            HbaseUtil.putRow(hbaseConn,GmallConfig.HBASE_NAMESPACE,sinkTable,rowKeyValue,sinkFamily,columnNames,columnValues);
        }
        //维度数据发生了变化，将Redis中缓存的数据清除掉
        if("delete".equals(type)||"update".equals(type)){
            DimUtil.deleteCached(sinkTable,rowKeyValue);
        }
    }
}
